package org.shj.spark.sql

import org.apache.spark.sql.SparkSession
import org.shj.spark.util.Util

object DataFrameDemo {
  def main(args: Array[String]): Unit = {
    val ss = SparkSession.builder().appName("DataFrameDemo").master("local").getOrCreate()
    ss.sparkContext.setLogLevel("WARN")
    
    import ss.implicits._
    
    val df = ss.read.json(Util.fullPath("employees.json"))
//    
//    df.printSchema()
//    
//    println("======== select(cols) ========")
//    df.select('name, 'salary + 1).show() // '加上字段名，可通过上面导入的隐式转换，转换成 Column
//    
//    println("======== filter(salary > 3500) ========")
//    df.filter("salary > 3500").show()
//    
//    println("======== cube(cols) ========")
//    df.cube("dept").avg().show()  // 不知道为什么会多一行 dept 为 null???
        
    /**
     * 注册成为GlobalTempView时，使用时，需加上 global_temp. 的前辍
     */
    df.createGlobalTempView("employee")
    
    val sql = "select * from global_temp.employee where salary > 3500"
    val empRdd = ss.sql(sql).filter("salary > 3500000").filter("name='aa'").rdd.map(row => {
      (row.getAs[String]("deptCd"), 1)
    })
    empRdd.reduceByKey(_+_).foreach(println)
    //empRdd.foreach(f)
    //empRdd.foreach(row => println(row.getAs("name")))
    
    /*println("====================== after a newSession() =============")
    //GlobalTempView在创建一个新的Session时，仍然可用
    val staffDf = ss.newSession().sql(sql)
    staffDf.show()
    
    println("============= implicit Encoder  =============")
    implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]]
    
    // row.getValuesMap[T] 一次性获取多个column，放到 Map[String, T] 中
    val ds = staffDf.map(staff => staff.getValuesMap[Any](List("name", "salary")))
    ds.collect().foreach(println)*/
    
    ss.stop()
  }
}